package com.bdc.Phoenix;

import com.alibaba.fastjson.JSONObject;
import com.bdc.Config.MyConfig;
import com.bdc.Util.PhoenixJDBCUtils;
import com.bdc.Util.RedisUtil;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;
import org.quartz.Job;
import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;
import redis.clients.jedis.Jedis;

import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.Properties;

/**
 * 查询Phoenix的历史数据，生成进Kafka的历史Topic主题中
 * Topic=LKYW_GPS_DATA_HISTORY
 */
public class QueryPhoenixToKafka implements Job {

    @Override
    public void execute(JobExecutionContext jobExecutionContext) throws JobExecutionException {
        RedisUtil redisUtil = new RedisUtil();
        String brokerList = MyConfig.getString("brokerList");
        Properties props = new Properties();
        props.put("bootstrap.servers", brokerList);
        props.put("acks", "all");
        props.put("key.serializer", StringSerializer.class.getName());
        props.put("value.serializer", StringSerializer.class.getName());
        KafkaProducer<String, String> producer = new KafkaProducer<>(props);
        Jedis jedis = redisUtil.getJedis();
        DateTimeFormatter df = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
        if (jedis.get("lkywCalTime") != null) {
            LocalDateTime localDateTime = LocalDateTime.parse(jedis.get("lkywCalTime"), df);
            String plusHours = localDateTime.plusHours(1).format(df);
            jedis.set("lkywCalTime", plusHours);
            //查询Phoenix数据
            Connection connection = PhoenixJDBCUtils.connect();
            String sql = "select * from GPS_DATA where date between '" + localDateTime.format(df) + "' and '" + plusHours + "' order by date asc";
            ResultSet resultSet = PhoenixJDBCUtils.query(sql);
            while (true) {
                try {
                    if (!resultSet.next()) break;
                    JSONObject json = new JSONObject();
                    json.put("vehicleno", resultSet.getString(1));
                    json.put("vehiclecolor", resultSet.getString(2));
                    json.put("date", resultSet.getString(3));
                    json.put("lon", resultSet.getString(4));
                    producer.send(new ProducerRecord<>("GPS_DATA_HISTORY", json.toJSONString()));
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }
            try {
                resultSet.close();
                connection.close();
            } catch (SQLException e) {
                e.printStackTrace();
            }
        }
        producer.close();
    }
}
